package com.danan.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.danan.realtime.app.BaseApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.pojo.TableProcess;
import com.danan.realtime.util.JDBCUtil;
import com.danan.realtime.util.SinkUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

/**
 * @author NanHuang
 * @Date 2023/1/25
 */
public class DwdApp_8_base_db extends BaseApp {
    public static void main(String[] args) {
        new DwdApp_8_base_db().init(4008,3,"DwdApp_8_base_db", Constant.TOPIC_ODS_DB);
    }
    @Override
    public void invoke(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        // 1 对数据进行ETL
        SingleOutputStreamOperator<JSONObject> etledDs = etl(ds);
        // 2 使用Flink CDC读取配置表的信息
        SingleOutputStreamOperator<TableProcess> dwdConfigDs = readDwdConfig(env);
        // 3 匹配数据和配置信息
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataAndConfigDs = matchDataAndConfig(etledDs, dwdConfigDs);
        // 4 将数据写入到Kafka中
        dataAndConfigDs.addSink(SinkUtil.getKafkaSink());
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> matchDataAndConfig(SingleOutputStreamOperator<JSONObject> data, SingleOutputStreamOperator<TableProcess> config) {
        // 1 将配置流设置为广播状态
        MapStateDescriptor<String, TableProcess> desc = new MapStateDescriptor<>("dwdConfig", String.class, TableProcess.class);
        BroadcastStream<TableProcess> bs = config.broadcast(desc);
        // 2 匹配数据和配置
        return data.connect(bs).process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject,TableProcess>>() {

            private Connection conn;
            // 解决配置信息迟到问题：预加载
            private HashMap<String, TableProcess> configState;


            @Override
            public void open(Configuration parameters) throws Exception {
                // 1 初始化状态
                configState = new HashMap<>();
                // 2 获取MySQL连接
                conn = JDBCUtil.getMySQLConnection();
                // 3 查询配置信息
                List<TableProcess> dwdConfig = JDBCUtil.queryList(conn, "select * from table_process where sink_type = ?", TableProcess.class, true, "dwd");
                // 4 状态赋值
                for (TableProcess value : dwdConfig) {
                    String key = String.format("%s:%s:%s",
                            value.getSourceTable(),
                            value.getSourceType(),
                            value.getSinkExtend() == null ? "" : value.getSinkExtend());
                    configState.put(key,value);
                }
                // 5 归还连接
                conn.close();
            }

            @Override
            public void processElement(JSONObject in, ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                // 1 获取广播状态
                ReadOnlyBroadcastState<String, TableProcess> bs = ctx.getBroadcastState(desc);
                // 2 拼接key
                String table = in.getString("table");
                String type = in.getString("type");
                String key = table + ":" + type + ":";
                JSONObject old = in.getJSONObject("old");
                JSONObject data = in.getJSONObject("data");
                if ("coupon_use".equals(table) && "update".equals(type)){
                    if ("1401".equals(old.getString("coupon_status"))
                        && "1402".equals(data.getString("coupon_status"))){
                        key += "{\"data\": {\"coupon_status\": \"1402\"}, \"old\": {\"coupon_status\": \"1401\"}}";
                    }
                    if (data.getString("used_time") != null){
                        key += "{\"data\": {\"used_time\": \"not null\"}}";
                    }
                }
                // 3 获取配置(先获取流，再获取预加载)
                TableProcess tp = bs.get(key);
                if (tp == null) {
                    tp = configState.get(key);
                }

                if (tp != null) {
                    // 4 删除流中不需要的字段信息
                    List<String> columns = Arrays.asList(tp.getSinkColumns().split(","));
                    data.keySet().removeIf(k -> !columns.contains(k));
                    // 5 返回结果
                    out.collect(new Tuple2<>(data,tp));
                }
            }

            @Override
            public void processBroadcastElement(TableProcess value, Context ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                // 1 获取广播状态
                BroadcastState<String, TableProcess> bs = ctx.getBroadcastState(desc);
                // 2 更新广播状态
                String key = String.format("%s:%s:%s",
                        value.getSourceTable(),
                        value.getSourceType(),
                        value.getSinkExtend() == null ? "" : value.getSinkExtend());
                bs.put(key,value);
            }
        });
    }

    private SingleOutputStreamOperator<TableProcess> readDwdConfig(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall_config")
                .tableList("gmall_config.table_process")
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        return env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"mysql_cdc")
                .map(JSON::parseObject)
                .process(new ProcessFunction<JSONObject, TableProcess>() {
                    @Override
                    public void processElement(JSONObject in, ProcessFunction<JSONObject, TableProcess>.Context ctx, Collector<TableProcess> out) throws Exception {
                        String op = in.getString("op");
                        TableProcess tp = null;
                        if ("d".equals(op)) {
                            tp = JSON.parseObject(in.getString("before"),TableProcess.class);
                        } else {
                            tp = JSON.parseObject(in.getString("after"),TableProcess.class);
                        }
                        tp.setOp(op);
                        if ("dwd".equals(tp.getSinkType())) {
                            out.collect(tp);
                        }
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> ds) {
        return ds.map(s -> s.replaceAll("bootstrap-",""))
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        try {
                            JSONObject obj = JSON.parseObject(s);
                            return "gmall2023".equals(obj.getString("database"))
                                    && obj.getString("table") != null
                                    && ("insert".equals(obj.getString("type")) || "update".equals(obj.getString("type")))
                                    && obj.getJSONObject("data") != null;
                        } catch (Exception e) {
                            System.out.println("数据格式错误，无法解析！（" + s + "）");
                            return false;
                        }
                    }
                })
                .map(JSON::parseObject);
    }
}
